package com.kitten.note.biz.consumer;

import com.alibaba.nacos.shaded.com.google.common.util.concurrent.RateLimiter;
import com.kitten.framework.common.utils.JsonUtils;
import com.kitten.note.biz.constants.MQConstants;
import com.kitten.note.biz.domain.dataobject.NoteCollectionDO;
import com.kitten.note.biz.domain.mapper.NoteCollectionDOMapper;
import com.kitten.note.biz.model.dto.CollectUnCollectNoteMqDTO;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.Objects;

/**
 * @author kitten
 */
@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "kittenBook_group_" + MQConstants.TOPIC_COLLECT_OR_UNCOLLECT,
        topic = MQConstants.TOPIC_COLLECT_OR_UNCOLLECT,
        consumeMode = ConsumeMode.ORDERLY
)
public class CollectUnCollectNoteConsumer implements RocketMQListener<Message> {

    @Resource
    private NoteCollectionDOMapper noteCollectionDOMapper;

    private RateLimiter rateLimiter = RateLimiter.create(1000);
    @Autowired
    private RocketMQTemplate rocketMQTemplate;


    @Override
    public void onMessage(Message message) {
        rateLimiter.acquire();

        String bodyStr = new String(message.getBody());
        String tag = message.getTags();
        log.info("CollectUnCollectNoteConsumer received message: {}, tag: {}", bodyStr, tag);
        if (Objects.equals(tag, MQConstants.TAG_COLLECT)) {
            handleCollectNoteTagMessage(bodyStr);
        } else if (Objects.equals(tag, MQConstants.TAG_UNCOLLECT)) {
            handleUnCollectNoteTagMessage(bodyStr);
        }
    }

    /**
     * 处理收藏笔记的消息
     * @param bodyStr
     */
    private void handleCollectNoteTagMessage(String bodyStr) {
        CollectUnCollectNoteMqDTO dto = JsonUtils.parseObject(bodyStr, CollectUnCollectNoteMqDTO.class);
        // 用户ID
        Long userId = dto.getUserId();
        // 收藏的笔记ID
        Long noteId = dto.getNoteId();
        // 操作类型
        Integer type = dto.getType();
        // 收藏时间
        LocalDateTime createTime = dto.getCreateTime();

        // 构建 DO 对象
        NoteCollectionDO noteCollectionDO = NoteCollectionDO.builder()
                .userId(userId)
                .noteId(noteId)
                .createTime(createTime)
                .status(type)
                .build();

        // 添加或更新笔记收藏记录
        int count = noteCollectionDOMapper.insertOrUpdate(noteCollectionDO);

        // MQ 计数服务
        org.springframework.messaging.Message<String> message = MessageBuilder.withPayload(bodyStr).build();
        String topic = MQConstants.TOPIC_COUNT_NOTE_COLLECT;
        rocketMQTemplate.asyncSend(topic, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("MQ 笔记收藏计数成功发送消息: {} ", bodyStr);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("MQ 笔记收藏计数发送消息失败: {}", throwable);
            }
        });
    }

    /**
     * 处理取消收藏笔记的消息
     * @param bodyStr
     */
    private void handleUnCollectNoteTagMessage(String bodyStr) {
        CollectUnCollectNoteMqDTO dto = JsonUtils.parseObject(bodyStr, CollectUnCollectNoteMqDTO.class);
        // 用户ID
        Long userId = dto.getUserId();
        // 收藏的笔记ID
        Long noteId = dto.getNoteId();
        // 操作类型
        Integer type = dto.getType();
        // 收藏时间
        LocalDateTime createTime = dto.getCreateTime();

        // 构建 DO 对象
        NoteCollectionDO noteCollectionDO = NoteCollectionDO.builder()
                .userId(userId)
                .noteId(noteId)
                .createTime(createTime)
                .status(type)
                .build();
        noteCollectionDOMapper.update2UnCollectByUserIdAndNoteId(noteCollectionDO);

        // TODO: MQ 计数服务
        org.springframework.messaging.Message<String> message = MessageBuilder.withPayload(bodyStr).build();
        String topic = MQConstants.TOPIC_COUNT_NOTE_COLLECT;
        rocketMQTemplate.asyncSend(topic, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("MQ 笔记收藏计数成功发送消息: {} ", bodyStr);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("MQ 笔记收藏计数发送消息失败: {}", throwable);
            }
        });

    }
}
